package com.sn.flink.streaming

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

object SocketWindowWordCountScala {

  def main(args: Array[String]): Unit = {
    val port: Int = try {
      ParameterTool.fromArgs(args).getInt("port")
    } catch {
      case e: Exception => {
        System.err.println("No specified. use default 9000")
      }
        9000
    }

    //获取运行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //连接此socket获取数据
    val text = env.socketTextStream("spark1", port, '\n')

    //解析数据，分组，窗口化，并且聚合求SUM
    import org.apache.flink.api.scala._

    val windowCounts = text
      .flatMap(line => line.split("\\s"))
      .map(w => WordWithCount(w, 1))
      .keyBy("word")
      .timeWindow(Time.seconds(2), Time.seconds(1)) //指定窗口大小，指定间隔时间
      .sum("count") //sum或者reduce都可以
    //.reduce((a,b) => WordWithCount(a.word,a.count+b.count))

    //使用一个单线程来打印结果
    windowCounts.print().setParallelism(1)

    //执行任务
    env.execute("Socket Window WordCount")
  }

  case class WordWithCount(word: String, count: Long)

}
